package com.persagy.iot.func

import com.persagy.iot.bean.IOTData
import com.persagy.iot.utils.HbaseUtil
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes

import scala.util.Try

/**
 * 分精度0数据写入 HBase
 */
class IOTAccuracySinkFunction extends RichSinkFunction[IOTData] {

  val accuracy_tableName: String = "accuracy_data"
  val family: String = "data"
  val sysTime: String = "sys_time"
  val value: String = "value"

  var conn: Connection = _
  var table: Table = _

  override def open(parameters: Configuration): Unit = {
    println("init Hbase ......")
    conn = HbaseUtil.getHbaseConnection()
    table = conn.getTable(TableName.valueOf(accuracy_tableName))
  }

  /**
   * 查询数据，对比时间戳，再修改
   */
  override def invoke(iotData: IOTData, context: SinkFunction.Context[_]): Unit = {
    val put = new Put(Bytes.toBytes(iotData.rowKey))
    println(iotData.sysTime)
    println(iotData.value)
    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(sysTime), Bytes.toBytes(iotData.sysTime.toString))
    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(value), Bytes.toBytes(iotData.value.toString))
    Try(table.put(put))
  }

  override def close(): Unit = {
    try {
      if (table != null) {
        table.close()
      }
      if (conn != null) {
        conn.close()
      }
      println("init Hbase create connection ")
    } catch {
      case e:Exception => System.err.println(e.getMessage)
    }
    println("close Hbase 数据。。。")
  }

  def now(): Long ={
    System.currentTimeMillis()
  }
}
